package qq2564874169.tuaotuao.fx.config.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import qq2564874169.tuaotuao.fx.FxAssert;
import qq2564874169.tuaotuao.fx.FxStream;
import qq2564874169.tuaotuao.fx.StringKeyMap;
import qq2564874169.tuaotuao.fx.config.ConfigReader;
import qq2564874169.tuaotuao.fx.utils.ExUtil;
import qq2564874169.tuaotuao.fx.utils.StringUtil;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class ZkReader implements ConfigReader {
    private static String rootNode;
    private ZooKeeper zk;
    private StringKeyMap<String> data = new StringKeyMap<>();

    public static void createRootNode(ZooKeeper zk) {
        createRootNode(zk, "/FxConfig");
    }

    public static void createRootNode(ZooKeeper zk, String nodeName) {
        try {
            if (zk.exists(nodeName, false) == null) {
                zk.create(nodeName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            rootNode = nodeName;
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    public ZkReader(ZooKeeper zk) {
        this.zk = zk;
        try {
            load();
            watch();
        } catch (InterruptedException | KeeperException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    private void load() throws InterruptedException, KeeperException {
        var names = zk.getChildren(rootNode, false);
        for (var name : names) {
            var kslist = zk.getChildren(fullPath(name), false);
            for (var k : kslist) {
                var buf = zk.getData(fullPath(name + "/" + k), false, new Stat());
                var key = name + "." + k;
                var value = new String(buf, StandardCharsets.UTF_8);
                data.put(key, value);
            }
        }
    }

    private void watch() throws InterruptedException, KeeperException {
        zk.addWatch(rootNode, e -> {
            if (e.getState().equals(Watcher.Event.KeeperState.Closed)) {
                return;
            }
            try {
                var path = e.getPath().substring(rootNode.length() + 1);
                var idx = path.indexOf("/");
                var namespace = "";
                var name = "";
                if (idx < 0) {
                    namespace = path;
                } else if (idx > 0) {
                    namespace = path.substring(0, idx);
                    name = path.substring(idx + 1);
                }
                switch (e.getType()) {
                    case NodeCreated:
                    case NodeDataChanged:
                        if (!name.isEmpty()) {
                            var buf = zk.getData(e.getPath(), false, new Stat());
                            var value = new String(buf, StandardCharsets.UTF_8);
                            data.put(namespace + "." + name, value);
                        }
                        break;
                    case NodeDeleted:
                        if (name.isEmpty() && !namespace.isEmpty()) {
                            var pre = namespace + ".";
                            FxStream.create(data.keySet())
                                    .filter(i -> i.startsWith(pre))
                                    .foreach(k -> data.remove(k));
                        } else if (!name.isEmpty()) {
                            data.remove(namespace + "." + name);
                        }
                        break;
                }
            } catch (KeeperException | InterruptedException ex) {
                throw ExUtil.toRuntime(ex);
            }
        }, AddWatchMode.PERSISTENT_RECURSIVE);
    }

    private static String fullPath(String path) {
        FxAssert.isNotNull(path);
        FxAssert.isNotNull(rootNode, "请先调用createRootNode创建根节点");
        if (path.startsWith("/")) {
            return rootNode + path;
        } else {
            return rootNode + "/" + path;
        }
    }

    @Override
    public String read(String namespace, String name) {
        if (namespace != null && !namespace.isEmpty()) {
            name = namespace + "." + name;
        }
        name = name.toLowerCase();
        if (data.containsKey(name)) {
            return data.get(name);
        }
        return null;
    }

    @Override
    public List<String> getNames(String namespace) {
        if (StringUtil.hasValue(namespace)) {
            String find = namespace.toLowerCase() + ".";
            return FxStream.create(data.keySet()).filter(i -> i.startsWith(find)).map(i -> i.substring(find.length())).toList();
        } else {
            return new ArrayList<>();
        }
    }

    @Override
    public void reload() {
        data.clear();
        try {
            load();
        } catch (InterruptedException | KeeperException e) {
            throw ExUtil.toRuntime(e);
        }
    }
}
